package ru.nagakhl.whirlpool.pool;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import ru.nagakhl.whirlpool.exceptions.PoolException;

/**
 * Simple pool of connections to one cassandra host. Implementation is thread
 * safe with blocking support.
 * 
 * This pool doesn't support cleaning dirty connections or reinitializing them,
 * so use with care.
 * 
 * Also no close methods are provided, so connections should be manually closed
 * by user of this pool.
 * 
 * @author nagakhl
 * 
 */
public class SimplePool implements BlockingConnectionPool {

	protected final int initial;

	protected final int max;

	protected final int port;

	protected int index = -1;

	protected int count = 0;

	protected final String host;

	protected final Logger log;

	protected final PooledClient[] clients;

	protected final Lock lock;

	protected final Condition notEmpty;

	/**
	 * Construct pool with default capacity and connection parameters Initial
	 * connection count is 1 Maximum connection count is 10 Connections are
	 * established to localhost on port 9160
	 */
	public SimplePool() {
		this(1, 10, "127.0.0.1", 9160);
	}

	/**
	 * Construct pool with default connection parameters Connections are
	 * established to localhost on port 9160
	 * 
	 * @param initial
	 *            Initial count of connections this pool must initialize
	 * 
	 * @param max
	 *            Maximum count of connections this pool will contain if more
	 *            connections are requested then calling thread will be blocked
	 *            until at least one connection not returned to pool
	 */
	public SimplePool(int initial, int max) {
		this(initial, max, "127.0.0.1", 9160);
	}

	/**
	 * Construct connection pool with given params
	 * 
	 * @param initial
	 *            Initial count of connections this pool must initialize
	 * 
	 * @param max
	 *            Maximum count of connections this pool will contain if more
	 *            connections are requested then calling thread will be blocked
	 *            until at least one connection not returned to pool
	 * 
	 * @param host
	 *            Host that underlying thrift connection must be established to
	 * 
	 * @param port
	 *            Port that underlying thrift connection must be established to
	 */
	public SimplePool(int initial, int max, String host, int port) {
		this.initial = initial;
		this.max = max;
		this.host = host;
		this.port = port;
		log = Logger.getLogger(getClass().getSimpleName() + "->" + host + ":" + port);
		clients = new PooledClient[max];
		lock = new ReentrantLock();
		notEmpty = lock.newCondition();
		try {
			for (int i = 0; i < initial; i++) {
				clients[i] = initConnection();
				++index;
				++count;
			}
		} catch (TTransportException e) {
			log.log(Level.SEVERE, "Exception opening connection", e);
			throw new PoolException("Exception opening connection", e);
		}
	}

	/**
	 * Get connection from pool. If there are no connections in pool and max
	 * count is not yet reached then connection will be created and initialized.
	 * If maximum count is already reached then calling thread will be blocked
	 * until at least one connection is not returned to pool.
	 */
	public Cassandra.Client getConnection() {
		lock.lock();
		try {
			Cassandra.Client client;
			if (index < 0 && count < max) {
				client = initConnection();
				++count;
			} else {
				while (index < 0)
					notEmpty.await();
				client = clients[index];
				clients[index] = null;
				--index;
			}
			return client;
		} catch (TTransportException e) {
			String err = "Exception obtaining additional connection to database.";
			log.log(Level.SEVERE, err, e);
			throw new PoolException(err, e);
		} catch (InterruptedException e) {
			String err = "Interrupted while waiting for connection.";
			log.log(Level.SEVERE, err, e);
			throw new PoolException(err, e);
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Return connection to pool, notifying one of waiting for connection
	 * threads if there are some.
	 */
	public void returnConnection(Cassandra.Client cl) throws PoolException {
		if (cl instanceof PooledClient) {
			PooledClient client = (PooledClient) cl;
			if (client.getHome() == this) {
				lock.lock();
				try {
					clients[++index] = client;
					notEmpty.signal();
				} finally {
					lock.unlock();
				}
			} else {
				throw new PoolException("Attempt to return connection into wrong pool.");
			}
		} else {
			throw new PoolException("Attempt to return not pooled connection back to pool.");
		}
	}

	/**
	 * Tests if this pool has available connections. 
	 * 
	 * Don't rely on this method result in multithreaded environment, cause result may 
	 * change between this method invocation and execution of logic dependent on it's return value.
	 */
	public boolean hasConnectons() {
		return index < 0 ? false : true;
	}
	
	/**
	 * Get count of available connections. 
	 * 
	 * Don't rely on this method result in multithreaded environment, cause result may 
	 * change between this method invocation and execution of logic dependent on it's return value.
	 */
	public int available() {
		return index + 1;
	}
	
	/**
	 * Get count of all connections? initialized by current pool. 
	 * 
	 * Don't rely on this method result in multithreaded environment, cause result may 
	 * change between this method invocation and execution of logic dependent on it's return value.
	 */
	public int initialized() {
		return count;
	}
	
	protected PooledClient initConnection() throws TTransportException {
		TTransport transport = new TSocket(host, port);
		TProtocol proto = new TBinaryProtocol(transport);
		PooledClient client = new PooledClient(proto, this);
		transport.open();
		return client;
	}
}
